package preprocess;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import utils.Utils;

public class MappingACookieUserId  extends Configured implements Tool {

	public static String getACookieUserIdMapPath(String date) {
		return Utils.rootDir + date + "/ACookieUserIdMap/";
	}
	
	enum Counter{INVALID, NO_ID, TOTAL, MULTI_ID_COUNT, UNIQUE_ID_COUNT};
	public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		@Override 
		public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			String[] strs = value.toString().split("\u0001");
			reporter.incrCounter(Counter.TOTAL, 1);
			if (strs.length != 9 || !utils.Utils.isValid(strs[8])) {
				reporter.incrCounter(Counter.INVALID, 1);
				if (strs.length == 9)
					reporter.incrCounter(Counter.NO_ID, 1);
				return;				
			}				
			k.set(strs[2]);
			v.set(strs[8]);
			output.collect(k, v);
		}
	}
	
	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
		@Override
		public void reduce(Text key, Iterator<Text> iter, OutputCollector<Text, Text> output, Reporter reporter) throws IOException{
			Set<String> strSet = new HashSet<String>();
			while (iter.hasNext()) {
				strSet.add(iter.next().toString());				
			}
			if (strSet.size() != 1) {
				reporter.incrCounter(Counter.MULTI_ID_COUNT, 1);
				return;
			}
			reporter.incrCounter(Counter.UNIQUE_ID_COUNT, 1);
			output.collect(key, new Text(strSet.iterator().next()));
		}
	}
	
	/*
	 * (non-Javadoc)
	 * UserFiltered Table
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/ACookieUserIdMapping/
	 * 1.ACookie
	 * 2.UserId 数字Id
	 */
	@Override
	public int run(String[] args) throws Exception {
		String aCookieLogPrefix = "/group/tbads/logdata/taobao_acookie/" + args[0];
		String output = getACookieUserIdMapPath(args[0]);
		System.out.println("start job MappingAcookieUserId...");
		System.out.println("aCookieLogPrefix: " + aCookieLogPrefix);
		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		JobConf conf = new JobConf(getConf(), MappingACookieUserId.class);
		conf.setJobName("ACookieUserIdMappingJob");
		
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		
		conf.setMapperClass(Map.class);
		conf.setCombinerClass(Reduce.class);
		conf.setReducerClass(Reduce.class);
		
		conf.setOutputFormat(TextOutputFormat.class);
		String[] minStrs = {"00", "30"};
		String[] serverStrs = {"ttslave27.sds.cnz.alimama.com", 
				"ttslave28.sds.cnz.alimama.com", 
				"ttslave29.sds.cnz.alimama.com", 
				"ttslave30.sds.cnz.alimama.com"};
		for (int i = 0; i < 24; ++i) {
			String strHour;
			if (i < 10)
				strHour = "0" + String.valueOf(i);
			else 
				strHour = String.valueOf(i);
			for (String strMin : minStrs) {
				for (String server : serverStrs) {
					String input = aCookieLogPrefix + "/" + strHour + "/" + strMin + "/" + server + "/";
					MultipleInputs.addInputPath(conf, new Path(input), SequenceFileInputFormat.class);					
				}
			}
		}
		FileOutputFormat.setOutputPath(conf, new Path(output));
		JobClient.runJob(conf);
		return 0;		 	
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new MappingACookieUserId(), args);
		System.exit(res);
	}
}
